#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"

#include "mq_c_channel.hpp"
#include "mq_worker.hpp"

namespace bitmq{
    class MyConnection{
        public:
        using ptr=std::shared_ptr<MyConnection>;
        // muduo::CountDownLatch _latch;
        // muduo::net::TcpClient _client;
        // ProtobufDispatcher _dispatcher;
        // ProtobufCodecPtr _codec;
        // muduo::net::TcpConnectionPtr _conn;
        // ChannelManager::ptr _cmanager;
        // AsyncWorker::ptr _worker;
        MyConnection(const std::string& sip,int sport,AsyncWorker::ptr& worker)
        :_latch(1)
        ,_client(worker->loopthread.startLoop()
        , muduo::net::InetAddress(sip, sport), "Client")
        ,_dispatcher(std::bind(&MyConnection::onUnknownMessage,this,
        std::placeholders::_1,std::placeholders::_2,std::placeholders::_3))
        ,_codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage,
        &_dispatcher,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)))
        ,_worker(worker)
        ,_cmanager(std::make_shared<ChannelManager>())
        {
            //注册分发器
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&MyConnection::onBasicResponse,this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&MyConnection::onconsumeResponse,this,
                std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            //注册io事件回调函数
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage,_codec.get(),std::placeholders::_1,
            std::placeholders::_2,std::placeholders::_3));
            //注册连接事件回调函数
            _client.setConnectionCallback(std::bind(&MyConnection::onConnection,this,std::placeholders::_1));

            _client.connect();
            _latch.wait();//阻塞等待，直到连接建立成功
            DLOG("获取到连接");
        }
        ClientChannel::ptr openChannel(){
            ClientChannel::ptr channel = _cmanager->create(_conn, _codec);
            bool ret = channel->openChannel();
            if (ret == false) {
                DLOG("打开信道失败！");
                return ClientChannel::ptr();
            }
            return channel;
        }
        void closeChannel(const ClientChannel::ptr &channel){
            channel->closeChannel();
            _cmanager->remove(channel->cid());
        }
        private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr& conn, const MessagePtr& message, muduo::Timestamp) {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }
        void onConnection(const muduo::net::TcpConnectionPtr& conn){
            if(conn->connected()){
                _conn=conn;
                DLOG("与服务端的连接建立成功！");
                _latch.countDown();
            }else{
                DLOG("开始关闭连接!");
                //_worker->pool.stop();
                _conn.reset();
            }
        }
        void onBasicResponse(const muduo::net::TcpConnectionPtr& conn,
             const basicCommonResponsePtr& message, muduo::Timestamp){
                ClientChannel::ptr tmp_channel = _cmanager->get(message->cid());
                if(tmp_channel.get() == nullptr){
                    DLOG("处理回应消息时,未找到对应的信道");
                    return ;
                }
                tmp_channel->putbasicCommonResponse(message);

             }
        void onconsumeResponse(const muduo::net::TcpConnectionPtr& conn,
            const basicConsumeResponsePtr& message, muduo::Timestamp){
                 ClientChannel::ptr tmp_channel = _cmanager->get(message->cid());
                 if (tmp_channel.get() == nullptr){
                     DLOG("处理回应消息时,未找到对应的信道");
                     return;
                 }
                 DLOG("处理回应消息的信道是 %s",message->cid().c_str());
                 _worker->pool.push([tmp_channel,message](){
                     tmp_channel->consume(message);
                 });
            }
        private:
        muduo::CountDownLatch _latch;
        muduo::net::TcpClient _client;
        ProtobufDispatcher _dispatcher;
        ProtobufCodecPtr _codec;
        muduo::net::TcpConnectionPtr _conn;
        ChannelManager::ptr _cmanager;
        AsyncWorker::ptr _worker;
    };
}
#endif